package com.landtool.lanbase.elasticsearch;

import cn.hutool.core.date.DateUtil;
import org.apache.http.HttpHost;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.io.IOException;
import java.util.Map;

public class RestClientUtils {

    private static Logger log = LoggerFactory.getLogger(RestClientUtils.class);

    /**
     * 高阶Rest Client
     */
    private RestHighLevelClient client = null;
    /**
     * 低阶Rest Client
     */
    private RestClient restClient = null;

    private String ip = "192.168.0.25";

    private Integer port = 9222;


    /**
     * 这里使用饿汉单例模式创建RestHighLevelClient
     */
    public RestClientUtils() {
        if (client == null) {
            synchronized (RestHighLevelClient.class) {
                if (client == null) {
                    client = getClient();
                }
            }
        }
    }

    private RestHighLevelClient getClient() {
        RestHighLevelClient client = null;

        try {
            client = new RestHighLevelClient(RestClient.builder(new HttpHost(ip, port, "http")
                    , new HttpHost("192.168.0.28", 9222, "http")));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return client;
    }

    private RestClient getRestClient() {
        RestClient client = null;

        try {
            client = RestClient.builder(new HttpHost(ip, port, "http")
                    , new HttpHost("192.168.0.28", 9222, "http")).build();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return client;
    }

    public void closeClient() {
        try {
            if (client != null) {
                client.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void migrateData() {
        //输入开始的时间
        String dateStr = "2019-06-10 16:00:00";
        long startDate = DateUtil.parse(dateStr).getTime();
        //时间差 一个小时
        long timeInterval = 5 * 60 * 1000;

        //循环24次 就是一天
        int hour = 24 * 22;
        for (int i = 0; i < hour; i++) {
            long endData = startDate + timeInterval;
            //循环结束
            boolean isEnd=searchByScroll(startDate,  endData);
            if(isEnd){
                startDate += timeInterval;
            }
        }
    }

    private boolean searchByScroll(long startDate, long endData) {
        log.info("-----提交状态：开始提交------提交时间段：" + startDate + "---" + endData + "-------");
        RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("timestamp").gt(startDate).lt(endData);
        //要搜索的数据库
        SearchRequest searchRequest = new SearchRequest("his_2019_06");
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.size(2000);
        searchSourceBuilder.query(rangeQueryBuilder);
        searchRequest.source(searchSourceBuilder);
        searchRequest.scroll(new TimeValue(3000));

        SearchResponse searchResponse = null;
        try {
            searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        String scrollId = searchResponse.getScrollId();

        while (true){
            SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
            scrollRequest.scroll(TimeValue.timeValueSeconds(30));
            SearchResponse searchScrollResponse = null;
            try {
                searchScrollResponse = client.scroll(scrollRequest, RequestOptions.DEFAULT);
            } catch (IOException e) {
                e.printStackTrace();
            }
            scrollId = searchScrollResponse.getScrollId();
            SearchHits searchHits = searchScrollResponse.getHits();
            BulkRequest bulkRequest = new BulkRequest();
            for (SearchHit hit : searchHits) {
                Map <String, Object> sourceAsMap = hit.getSourceAsMap();
                //保存到 新的数据库
                bulkRequest.add(new IndexRequest("his_2019_07", "_doc").source(sourceAsMap));
            }
            long totalHits = searchHits.getHits().length;
            if (totalHits > 0) {
                try {
                    BulkResponse bulkResponses = client.bulk(bulkRequest, RequestOptions.DEFAULT);
                    log.info("-----提交总数:" + totalHits);
                    log.info("-----提交状态：" + bulkResponses.status() + "------提交时间段：" + startDate + "---" + endData + "-----");
                } catch (IOException e) {
                    e.printStackTrace();
                }
            } else {
                log.info("-----提交状态：该时间段以提交完成------时间间隔内在无数据");
                break;
            }
        }
        return true;
    }

}



